package com.ali.demo.unicom.ons;

import java.util.Properties;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

public class ONSConsumerBean {

    private String consumerId;
    private String accessKey;
    private String secretKey;

    private String topic;
    private String msgTag;
    private MessageListener listener;
    private int consumeThreads;

    private Consumer consumer;

    public String getConsumerId() {
	return consumerId;
    }

    public void setConsumerId(String consumerId) {
	this.consumerId = consumerId;
    }

    public String getAccessKey() {
	return accessKey;
    }

    public void setAccessKey(String accessKey) {
	this.accessKey = accessKey;
    }

    public String getSecretKey() {
	return secretKey;
    }

    public void setSecretKey(String secretKey) {
	this.secretKey = secretKey;
    }

    public String getTopic() {
	return topic;
    }

    public void setTopic(String topic) {
	this.topic = topic;
    }

    public String getMsgTag() {
	return msgTag;
    }

    public void setMsgTag(String msgTag) {
	this.msgTag = msgTag;
    }

    public MessageListener getListener() {
	return listener;
    }

    public void setListener(MessageListener listener) {
	this.listener = listener;
    }

    public int getConsumeThreads() {
	return consumeThreads;
    }

    public void setConsumeThreads(int consumeThreads) {
	this.consumeThreads = consumeThreads;
    }

    public Consumer getConsumer() {
	return consumer;
    }

    public void init() {
	Properties properties = new Properties();
	properties.put(PropertyKeyConst.ConsumerId, consumerId);
	properties.put(PropertyKeyConst.AccessKey, accessKey);
	properties.put(PropertyKeyConst.SecretKey, secretKey);

	if (consumeThreads > 0) {
	    properties.put(PropertyKeyConst.ConsumeThreadNums,
		    Integer.toString(consumeThreads));
	}

	consumer = ONSFactory.createConsumer(properties);
	consumer.subscribe(topic, (msgTag == null) ? "*" : msgTag, listener);
	consumer.start();
    }

    public void close() {
	if (consumer != null) {
	    consumer.shutdown();
	}
    }
}
